﻿using System;
using System.Configuration;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using RabbitMQ.Client;
using Newtonsoft.Json;
using Rongzi.Infrastructure.ServiceBus.Config;
using RabbitMQ.Client.Events;

namespace Rongzi.Infrastructure.ServiceBus.Queue
{
    public class RabbitMQProvider : IMessageQueueProvider, IDisposable
    {
        private static IConnection connection;
        private static RabbitMQConfigurationElement rabbitMQConfig;
        private static RabbitMQProvider provider;
        private static object asyncRoot = new object();

        public RabbitMQProvider()
        {

        }

        public static RabbitMQProvider Instance
        {
            get
            {
                if (provider == null)
                {
                    lock (asyncRoot)
                    {
                        if (provider == null)
                        {
                            provider = new RabbitMQProvider();
                        }
                    }
                }

                return provider;
            }
        }

        static RabbitMQProvider()
        {
            MessageQueueConfigSection messageQueueConfigSection = ConfigurationManager.GetSection("queueConfig") as MessageQueueConfigSection;
            if (messageQueueConfigSection == null)
            {
                throw new ArgumentNullException("queueConfig section is missing.");
            }
            rabbitMQConfig = messageQueueConfigSection.RabbitMQ;
            if (rabbitMQConfig == null)
            {
                throw new ArgumentNullException("rabbitMQ ConfigurationElement is missing.");
            }
            if (!rabbitMQConfig.Enable)
                return;

            InitConnection();
        }

        private static void InitConnection()
        {
            if (connection != null)
                return;

            var factory = new ConnectionFactory()
            {
                UserName = rabbitMQConfig.UserName,
                Password = rabbitMQConfig.Password,
                AutomaticRecoveryEnabled = true,
                TopologyRecoveryEnabled = true,
                RequestedHeartbeat = 60
            };
            if (!string.IsNullOrEmpty(rabbitMQConfig.VirtualHost))
            {
                factory.VirtualHost = rabbitMQConfig.VirtualHost;
            }
            if (rabbitMQConfig.Port.GetValueOrDefault() > 0)
            {
                factory.Port = rabbitMQConfig.Port.Value;
            }
            try
            {
                connection = factory.CreateConnection(rabbitMQConfig.HostName.Split(new string[] { "," }, StringSplitOptions.RemoveEmptyEntries));
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }

        public void Publish<T>(T message, bool durable = true) where T : class
        {
            if (!rabbitMQConfig.Enable || connection == null || message == null)
                return;

            using (var channel = connection.CreateModel())
            {
                var exchange = typeof(T).FullName;
                channel.ExchangeDeclare(exchange: exchange, type: "fanout", durable: durable);

                var body = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(message));
                channel.BasicPublish(exchange: exchange, routingKey: "", basicProperties: null, body: body);
            }
        }

        /// <summary>
        /// 订阅消息，自动应答
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="subscriptionId">队列ID</param>
        /// <param name="onMessage">消息处理程序</param>
        /// <param name="durable">是否持久化</param>
        /// <returns></returns>

        public bool Subscribe<T>(string subscriptionId, Action<T> onMessage, bool durable = true) where T : class
        {
            if (!rabbitMQConfig.Enable || connection == null)
                return false;

            var channel = connection.CreateModel();

            var exchange = typeof(T).FullName;
            channel.ExchangeDeclare(exchange: exchange, type: "fanout", durable: durable);

            string queueName = "";
            if (string.IsNullOrEmpty(subscriptionId))
            {
                queueName = channel.QueueDeclare().QueueName;
            }
            else
            {
                queueName = $"{typeof(T).FullName}:{subscriptionId}";
            }
            if (!string.IsNullOrEmpty(queueName))
            {
                channel.QueueDeclare(queue: queueName, durable: durable, exclusive: false, autoDelete: false, arguments: null);
            }

            channel.QueueBind(queue: queueName, exchange: exchange, routingKey: "");

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var message = Encoding.UTF8.GetString(ea.Body);
                onMessage(JsonConvert.DeserializeObject<T>(message));

            };
            channel.BasicConsume(queue: queueName, noAck: true, consumer: consumer);
            return true;
        }

        /// <summary>
        /// 订阅消息，需要手动应答
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="subscriptionId">队列ID</param>
        /// <param name="handleMessage">消息处理程序，返回是否处理成功</param>
        /// <param name="durable">是否持久化</param>
        /// <returns></returns>
        public bool SubscribeAck<T>(string subscriptionId, Func<T, bool> handleMessage, bool durable = true) where T : class
        {
            if (!rabbitMQConfig.Enable || connection == null)
                return false;

            var channel = connection.CreateModel();

            var exchange = typeof(T).FullName;
            channel.ExchangeDeclare(exchange: exchange, type: "fanout", durable: durable);

            string queueName = "";
            if (string.IsNullOrEmpty(subscriptionId))
            {
                queueName = channel.QueueDeclare().QueueName;
            }
            else
            {
                queueName = $"{typeof(T).FullName}:{subscriptionId}";
            }
            if (!string.IsNullOrEmpty(queueName))
            {
                channel.QueueDeclare(queue: queueName, durable: durable, exclusive: false, autoDelete: false, arguments: null);
            }

            channel.QueueBind(queue: queueName, exchange: exchange, routingKey: "");

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                try
                {
                    var message = Encoding.UTF8.GetString(ea.Body);
                    if (handleMessage(JsonConvert.DeserializeObject<T>(message)))
                    {
                        channel.BasicAck(ea.DeliveryTag, false);
                    }
                    else
                    {
                        channel.BasicNack(ea.DeliveryTag, false, false);
                    }
                }
                catch (Exception ex)
                {
                    channel.BasicNack(ea.DeliveryTag, false, false);
                    throw ex;
                }
            };
            channel.BasicConsume(queue: queueName, noAck: false, consumer: consumer);
            return true;
        }

        #region delay message
        /// <summary>
        /// 发布延迟消息
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="message"></param>
        /// <param name="expireMinutes"></param>
        /// <param name="durable"></param>
        public void PublishDelayMessage<T>(T message, int expireMinutes, bool durable = true) where T : class
        {
            if (expireMinutes <= 0)
            {
                throw new ArgumentException("expireMinutes 必须大于0");
            }

            using (var channel = connection.CreateModel())
            {
                var body = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(message));

                //创建默认的死信交换机
                string receiveExchangeName = $"{typeof(T).FullName}.DelayReceive";
                string receiveQueueName = $"{receiveExchangeName}.{expireMinutes}";
                channel.ExchangeDeclare(exchange: receiveExchangeName, type: "direct", durable: durable);

                string bufferExchange = $"{typeof(T).FullName}.DelayBuffer";
                string bufferQueueName = $"{bufferExchange}.{expireMinutes}";
                channel.ExchangeDeclare(exchange: bufferExchange, type: "direct", durable: durable);

                //创建消息缓冲队列，在这个队列里面实现消息的过期转发
                var properties = channel.CreateBasicProperties();
                properties.Expiration = (expireMinutes * 60000).ToString();
                Dictionary<string, object> arguments = new Dictionary<string, object>();
                arguments.Add("x-dead-letter-exchange", receiveExchangeName);
                arguments.Add("x-dead-letter-routing-key", receiveQueueName);
                channel.QueueDeclare(queue: bufferQueueName, durable: durable, exclusive: false, autoDelete: false, arguments: arguments);
                channel.QueueBind(queue: bufferQueueName, exchange: bufferExchange, routingKey: bufferQueueName);

                //这个队列用于消息在缓冲队列中过期后转发的目标队列
                channel.QueueDeclare(queue: receiveQueueName, durable: durable, exclusive: false, autoDelete: false, arguments: null);
                channel.QueueBind(queue: receiveQueueName, exchange: receiveExchangeName, routingKey: receiveQueueName);

                channel.BasicPublish(exchange: bufferExchange, routingKey: bufferQueueName, basicProperties: properties, body: body);
            }
        }

        /// <summary>
        /// 订阅延迟消息
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="onMessage"></param>
        /// <param name="expireMinutes"></param>
        /// <param name="durable"></param>
        /// <returns></returns>
        public bool SubscribeDelayMessage<T>(Action<T> onMessage, int expireMinutes, bool durable = true) where T : class
        {
            if (expireMinutes <= 0)
            {
                throw new ArgumentException("expireMinutes 必须大于0");
            }

            var channel = connection.CreateModel();

            string receiveExchangeName = $"{typeof(T).FullName}.DelayReceive";
            string receiveQueueName = $"{receiveExchangeName}.{expireMinutes}";
            channel.ExchangeDeclare(exchange: receiveExchangeName, type: "direct", durable: durable);

            channel.QueueDeclare(queue: receiveQueueName, durable: durable, exclusive: false, autoDelete: false, arguments: null);
            channel.QueueBind(queue: receiveQueueName, exchange: receiveExchangeName, routingKey: receiveQueueName);

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                try
                {
                    var message = Encoding.UTF8.GetString(ea.Body);
                    onMessage(JsonConvert.DeserializeObject<T>(message));
                }
                catch
                {

                }
            };
            channel.BasicConsume(queue: receiveQueueName, noAck: true, consumer: consumer);
            return true;
        }
        #endregion

        public void Dispose()
        {
            if (connection != null)
                connection.Close();
        }
    }
}
